package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.func.DimAsyncFunction;
import com.atguigu.gmall.realtime.bean.TradeSkuOrderBean;
import com.atguigu.gmall.realtime.util.DateFormatUtil;
import com.atguigu.gmall.realtime.util.MyClickhouseUtil;
import com.atguigu.gmall.realtime.util.MyKafkaUtil;
import com.atguigu.gmall.realtime.util.TimestampLtz3CompareUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

/**
 * @author Felix
 * @date 2022/10/8
 * 交易域：SKU粒度下单业务过程聚合统计
 * 需求：按照SKU维度分组，统计各窗口的独立用户数、订单数、原始金额、活动减免金额、优惠券减免金额和订单金额，并关联维度信息，并写到CK
 * 需要启动的进程
 *      zk、kafka、maxwell、clickhouse、hdfs、hbase、redis
 *      DwdTradeOrderPreProcess、DwdTradeOrderDetail、DwsTradeSkuOrderWindow
 * 开发流程
 *      基本环境准备
 *      检查点相关设置
 *      从kafka的下单主题(事实表)读取下单数据
 *      对读取的数据进行类型转换    jsonStr->jsonObj
 *      去重
 *          为什么会产生重复数据？
 *              因为在dwd层准备订单预处理表的时候，我们将订单明细、订单、订单明细活动、订单明细优惠券、字典表这5张表进行关联，
 *              其中在关联订单明细活动、订单明细优惠券的时候，我们使用的是左外连接，如果左表数据先来，右表数据后到，会生成如下数据
 *                  +I  左表  null
 *                  -D  左表  null
 *                  +I  左表  右表
 *              将上述生成的数据发送到kafka主题的话，会有三条消息，左表 null，null，左表 右表；
 *              如果使用kafka连接器从kafka主题中读取数据的话，null消息会被过滤掉，但是[左表 null]和[左表 右表]重复了
 *          思路：
 *              Flink的状态编程 + 定时器
 *              第一条数据来的时候，先放到状态中，同时注册定时器，再有重复数据进来，我们会用后进来的数据和状态中的数据进行时间的对比，
 *              将时间靠后的放到状态中保存起来，等定时器被触发执行的时候，就会将状态中时间大的写到下游，实现去重效果
 *          步骤：
 *              先按照唯一键进行分组----订单明细id
 *              去重
 *      对流中的数据类型进行转换        jsonObj->实体类对象
 *      判断当前这条数据是不是下单独立用户
 *          按照用户id进行分组
 *          使用Flink的状态编程，判断是否下单独立用户
 *      指定Watermark以及提取事件时间字段
 *      按照统计的维度sku进行分组
 *      开窗
 *      聚合
 *      和维度进行关联
 *          基本的维度关联
 *              准备工作1：
 *                  在PhoenixUtil工具类中，定义到phoenix表中查询数据方法 List<T> queryList(sql,clz)
 *              准备工作2：
 *                  封装专门查询维度的工具类DimUtil  JSONObject getDimInfoNoCache(tableName,Tuple2<String,String>... params)
 *              在DWS的主应用程序中，通过map算子，对流中的数据进行转换，实现维度关联------同步处理
 *         优化1：旁路缓存
 *              思路：先从缓存中查询维度数据，如果缓存中找到了对应的维度数据，那么直接将缓存中的维度数据返回(缓存命中)；
 *                  如果在缓存中没有找到对应的维度数据，发送请求到phoenix表中将维度数据查询出来，并且将查询出来的维度放到缓存中，方便后续使用
 *              缓存产品选择
 *                  Redis √             性能不错，维护性好
 *                  状态                 性能高，维护性差
 *              Redis相关分析
 *                  key：        dim:维度表表名:主键1_主键2
 *                  type:        String
 *                  expire:      1day
 *                  注意： 如果业务数据库维度表发生了变化，需要将缓存中的维度数据删除掉
 *          优化2：异步IO
 *              默认情况下，在使用map算子对流中数据进行处理的时候，使用的同步的方式，也就是说，处理完流中一条数据才会处理下一条
 *              如果想去提升处理能力的话，可以通过调大并行度方式，但是更大的并行度意味着更多的硬件资源，不能无限制的提升，所以考虑使用异步的方式对流中数据进行处理
 *              异步语法：
 *                  AsyncDataStream.[un]orderedWait(
 *                      流,
 *                      异步操作 implements AsyncFunction,
 *                      超时时间,
 *                      时间单位
 *                  )
 *                  class DimAsyncFunction extends RichAsyncFunction{
 *                      asyncInvoke(){
 *                          //开启多个线程发送异步请求，完成维度关联
 *                          //模板方法设计模式  在父类中定义了完成某一个功能的核心算法骨架(步骤)，将具体的实现延迟到子类中去完成，子类在不改变父类核心算法
 *                          骨架的前提下，每一个子类都可以有自己不同的实现。
 *                      }
 *                  }
 *      将关联的结果写到CK中
 */
public class DwsTradeSkuOrderWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //TODO 2.检查点相关设置(略)
        //TODO 3.从kafka的下单主题中读取数据
        //3.1 声明消费的主题以及消费者组
        String topic = "dwd_trade_order_detail";
        String groupId = "dws_trade_sku_order_group";
        //3.2 创建消费者对象
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        //3.3 消费数据  封装为流
        DataStreamSource<String> kafkaStrDS = env.addSource(kafkaConsumer);

        //TODO 4.对读取的数据进行类型转换   jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(JSON::parseObject);
        // {"create_time":"2022-10-08 15:26:52","sku_num":"2","split_original_amount":"80.0000","sku_id":"23",
        // "date_id":"2022-10-08","source_type_name":"促销活动","user_id":"1","province_id":"33","source_type_code":"2404",
        // "row_op_ts":"2022-10-08 07:26:52.937Z","sku_name":"十月稻田 辽河长粒香 东北大米 5kg","id":"177","source_id":"1",
        // "order_id":"83","split_total_amount":"80.0","ts":"1665214012"}
        // jsonObjDS.print(">>>>");

        //TODO 5.按照唯一键(订单明细id)进行分组
        KeyedStream<JSONObject, String> orderDetailIdKeyedDS = jsonObjDS.keyBy(jsonObj -> jsonObj.getString("id"));

        //TODO 6.使用Flink的状态编程 + 定时器 实现去重
        SingleOutputStreamOperator<JSONObject> distinctDS = orderDetailIdKeyedDS.process(
            new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                private ValueState<JSONObject> lastJsonObjState;

                @Override
                public void open(Configuration parameters) throws Exception {
                    lastJsonObjState
                        = getRuntimeContext().getState(new ValueStateDescriptor<JSONObject>("lastJsonObjState", JSONObject.class));
                }

                @Override
                public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                    JSONObject lastJsonObj = lastJsonObjState.value();
                    //判断状态中是否保存了流中的数据
                    if (lastJsonObj == null) {
                        //如果状态中数据为null，说明当前处理的数据是该订单明细对应的第一条数据，将其放到状态中
                        lastJsonObjState.update(jsonObj);
                        //注册定时器
                        long currentProcessingTime = ctx.timerService().currentProcessingTime();
                        ctx.timerService().registerProcessingTimeTimer(currentProcessingTime + 5000L);
                    } else {
                        //如果状态中数据不为null，说明当前订单明细重复了，比较当前处理这条数据时间和状态中数据时间
                        // "row_op_ts":"2022-10-08 07:26:52.93Z",
                        // "row_op_ts":"2022-10-08 07:26:52.937Z",
                        String lastRowOpTs = lastJsonObj.getString("row_op_ts");
                        String curRowOpTs = jsonObj.getString("row_op_ts");
                        if (TimestampLtz3CompareUtil.compare(lastRowOpTs, curRowOpTs) <= 0) {
                            lastJsonObjState.update(jsonObj);
                        }
                    }
                }

                @Override
                public void onTimer(long timestamp, OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
                    //定时器执行代码
                    JSONObject jsonObj = lastJsonObjState.value();
                    if (jsonObj != null) {
                        out.collect(jsonObj);
                    }
                    lastJsonObjState.clear();
                }
            }
        );
        //TODO 7.将jsonObj转换为实体类对象
        SingleOutputStreamOperator<TradeSkuOrderBean> orderBeanDS = distinctDS.map(
            new MapFunction<JSONObject, TradeSkuOrderBean>() {
                @Override
                public TradeSkuOrderBean map(JSONObject jsonObj) throws Exception {
                    // {"create_time":"2022-10-08 15:26:52","sku_num":"2","split_original_amount":"80.0000","sku_id":"23",
                    // "date_id":"2022-10-08","source_type_name":"促销活动","user_id":"1","province_id":"33","source_type_code":"2404",
                    // "row_op_ts":"2022-10-08 07:26:52.937Z","sku_name":"十月稻田 辽河长粒香 东北大米 5kg","id":"177","source_id":"1",
                    // "order_id":"83","split_total_amount":"80.0","ts":"1665214012"}
                    String skuId = jsonObj.getString("sku_id");
                    String userId = jsonObj.getString("user_id");
                    String orderId = jsonObj.getString("order_id");
                    Double originalAmount = jsonObj.getDouble("split_original_amount");
                    Double totalAmount = jsonObj.getDouble("split_total_amount");
                    Double activityAmount = jsonObj.getDouble("split_activity_amount") == null ? 0.0 : jsonObj.getDouble("split_activity_amount");
                    Double couponAmount = jsonObj.getDouble("split_coupon_amount") == null ? 0.0 : jsonObj.getDouble("split_coupon_amount");
                    Long ts = jsonObj.getLong("ts") * 1000;

                    TradeSkuOrderBean orderBean = TradeSkuOrderBean.builder()
                        .skuId(skuId)
                        .userId(userId)
                        .orderIdSet(new HashSet(Collections.singleton(orderId)))
                        .originalAmount(originalAmount)
                        .couponAmount(couponAmount)
                        .orderUuCount(0L)
                        .activityAmount(activityAmount)
                        .orderAmount(totalAmount)
                        .ts(ts)
                        .build();
                    return orderBean;
                }
            }
        );
        // orderBeanDS.print(">>>>");

        //TODO 8.按照用户id进行分组
        KeyedStream<TradeSkuOrderBean, String> userIdKeyedDS = orderBeanDS.keyBy(TradeSkuOrderBean::getUserId);
        //TODO 9.使用Flink状态编程判断是否是下单独立用户
        SingleOutputStreamOperator<TradeSkuOrderBean> uuOrderBeanDS = userIdKeyedDS.process(
            new KeyedProcessFunction<String, TradeSkuOrderBean, TradeSkuOrderBean>() {
                private ValueState<String> lastOrderDateState;

                @Override
                public void open(Configuration parameters) throws Exception {
                    ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<String>("lastOrderDateState", String.class);
                    valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                    lastOrderDateState = getRuntimeContext().getState(valueStateDescriptor);
                }

                @Override
                public void processElement(TradeSkuOrderBean orderBean, Context ctx, Collector<TradeSkuOrderBean> out) throws Exception {
                    //获取状态中的上次下单日期
                    String lastOrderDate = lastOrderDateState.value();
                    //获取今天的下单日期
                    String curOrderDate = DateFormatUtil.toDate(orderBean.getTs());

                    if (StringUtils.isEmpty(lastOrderDate) || !lastOrderDate.equals(curOrderDate)) {
                        orderBean.setOrderUuCount(1L);
                        lastOrderDateState.update(curOrderDate);
                    }
                    out.collect(orderBean);
                }
            }
        );
        // uuOrderBeanDS.print(">>>>");

        //TODO 10.指定Watermark以及提取事件时间字段
        SingleOutputStreamOperator<TradeSkuOrderBean> withWatermarkDS = uuOrderBeanDS.assignTimestampsAndWatermarks(
            WatermarkStrategy
                .<TradeSkuOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(
                    new SerializableTimestampAssigner<TradeSkuOrderBean>() {
                        @Override
                        public long extractTimestamp(TradeSkuOrderBean orderBean, long recordTimestamp) {
                            return orderBean.getTs();
                        }
                    }
                )
        );
        //TODO 11.按照sku维度进行分组
        KeyedStream<TradeSkuOrderBean, String> skuIdKeyedDS = withWatermarkDS.keyBy(TradeSkuOrderBean::getSkuId);
        //TODO 12.开窗
        WindowedStream<TradeSkuOrderBean, String, TimeWindow> windowDS = skuIdKeyedDS.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));
        //TODO 13.聚合
        SingleOutputStreamOperator<TradeSkuOrderBean> reduceDS = windowDS.reduce(
            new ReduceFunction<TradeSkuOrderBean>() {
                @Override
                public TradeSkuOrderBean reduce(TradeSkuOrderBean value1, TradeSkuOrderBean value2) throws Exception {
                    value1.getOrderIdSet().addAll(value2.getOrderIdSet());
                    value1.setOrderUuCount(value1.getOrderUuCount() + value2.getOrderUuCount());
                    value1.setOriginalAmount(value1.getOriginalAmount() + value2.getOriginalAmount());
                    value1.setActivityAmount(value1.getActivityAmount() + value2.getActivityAmount());
                    value1.setCouponAmount(value1.getCouponAmount() + value2.getCouponAmount());
                    value1.setOrderAmount(value1.getOrderAmount() + value2.getOrderAmount());
                    return value1;

                }
            },
            new WindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                @Override
                public void apply(String s, TimeWindow window, Iterable<TradeSkuOrderBean> input, Collector<TradeSkuOrderBean> out) throws Exception {
                    String stt = DateFormatUtil.toYmdHms(window.getStart());
                    String edt = DateFormatUtil.toYmdHms(window.getEnd());

                    for (TradeSkuOrderBean orderBean : input) {
                        orderBean.setStt(stt);
                        orderBean.setEdt(edt);
                        orderBean.setTs(System.currentTimeMillis());
                        orderBean.setOrderCount((long) orderBean.getOrderIdSet().size());
                        out.collect(orderBean);
                    }
                }
            }
        );
        // reduceDS.print(">>>");
        //TODO 14.和SKU维度进行关联
        /*
        //同步的方式进行维度关联
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = reduceDS.map(
            new MapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
                @Override
                public TradeSkuOrderBean map(TradeSkuOrderBean orderBean) throws Exception {
                    //获取要关联的维度主键
                    String skuId = orderBean.getSkuId();
                    //根据维度的主键 获取维度对象  ID,SPU_ID,PRICE,SKU_NAME,SKU_DESC,WEIGHT,TM_ID,CATEGORY3_ID,SKU_DEFAULT_IMG,IS_SALE,CREATE_TIME
                    JSONObject dimInfoJsonObj = DimUtil.getDimInfo("dim_sku_info",skuId);
                    //将json对象的属性补充到流中的对象上
                    orderBean.setSkuName(dimInfoJsonObj.getString("SKU_NAME"));
                    orderBean.setTrademarkId(dimInfoJsonObj.getString("TM_ID"));
                    orderBean.setCategory3Id(dimInfoJsonObj.getString("CATEGORY3_ID"));
                    orderBean.setSpuId(dimInfoJsonObj.getString("SPU_ID"));
                    return orderBean;
                }
            }
        );

        withSkuInfoDS.print(">>>>");*/
        /*
        // 将异步I/O操作应用于DataStream作为DataStream的一次转换操作
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = AsyncDataStream.unorderedWait(
            reduceDS,
            // 实现分发请求的AsyncFunction
            new AsyncFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
                @Override
                public void asyncInvoke(TradeSkuOrderBean orderBean, ResultFuture<TradeSkuOrderBean> resultFuture) throws Exception {
                    //开启多线程发送异步请求
                    ThreadPoolExecutor poolExecutor = ThreadPoolUtil.getInstance();
                    poolExecutor.submit(
                        new Runnable() {
                            @Override
                            public void run() {
                                //根据流中对象获取要关联的维度主键
                                String skuId = orderBean.getSkuId();
                                //根据维度的主键获取对象的维度对象
                                JSONObject dimInfoJsonObj = DimUtil.getDimInfo("dim_sku_info", skuId);
                                //将维度对象相关的属性补充到流中对象上
                                if(dimInfoJsonObj != null){
                                    orderBean.setSkuName(dimInfoJsonObj.getString("SKU_NAME"));
                                    orderBean.setTrademarkId(dimInfoJsonObj.getString("TM_ID"));
                                    orderBean.setCategory3Id(dimInfoJsonObj.getString("CATEGORY3_ID"));
                                    orderBean.setSpuId(dimInfoJsonObj.getString("SPU_ID"));
                                }
                                //获取数据库交互的结果并发送给ResultFuture的回调函数
                                resultFuture.complete(Collections.singleton(orderBean));
                            }
                        }
                    );
                }
            },
            60, TimeUnit.SECONDS
        );

        withSkuInfoDS.print(">>>>");*/

        SingleOutputStreamOperator<TradeSkuOrderBean> wihtSkuInfoDS = AsyncDataStream.unorderedWait(
            reduceDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_sku_info") {
                @Override
                public void join(TradeSkuOrderBean orderBean, JSONObject dimInfoJsonObj) {
                    orderBean.setSkuName(dimInfoJsonObj.getString("SKU_NAME"));
                    orderBean.setTrademarkId(dimInfoJsonObj.getString("TM_ID"));
                    orderBean.setCategory3Id(dimInfoJsonObj.getString("CATEGORY3_ID"));
                    orderBean.setSpuId(dimInfoJsonObj.getString("SPU_ID"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getSkuId();
                }
            },
            60, TimeUnit.SECONDS
        );
        // wihtSkuInfoDS.print(">>>>>");

        //TODO 15.和SPU维度进行关联
        SingleOutputStreamOperator<TradeSkuOrderBean> withSpuInfoDS = AsyncDataStream.unorderedWait(
            wihtSkuInfoDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_spu_info") {
                @Override
                public void join(TradeSkuOrderBean orderBean, JSONObject dimInfoJsonObj) {
                    // ID,SPU_NAME,DESCRIPTION,CATEGORY3_ID,TM_ID
                    orderBean.setSpuName(dimInfoJsonObj.getString("SPU_NAME"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getSpuId();
                }
            },
            60, TimeUnit.SECONDS
        );
        // withSpuInfoDS.print(">>>>");
        //TODO 16.和TM维度进行关联
        SingleOutputStreamOperator<TradeSkuOrderBean> withTmDS = AsyncDataStream.unorderedWait(
            withSpuInfoDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_trademark") {
                @Override
                public void join(TradeSkuOrderBean orderBean, JSONObject dimInfoJsonObj) {
                    // ID,TM_NAME
                    orderBean.setTrademarkName(dimInfoJsonObj.getString("TM_NAME"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getTrademarkId();
                }
            }, 60, TimeUnit.SECONDS
        );
        // withTmDS.print(">>>>");
        //TODO 17.和category3维度进行关联
        SingleOutputStreamOperator<TradeSkuOrderBean> withCategory3Stream = AsyncDataStream.unorderedWait(
            withTmDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category3".toUpperCase()) {
                @Override
                public void join(TradeSkuOrderBean javaBean, JSONObject jsonObj) {
                    javaBean.setCategory3Name(jsonObj.getString("name".toUpperCase()));
                    javaBean.setCategory2Id(jsonObj.getString("category2_id".toUpperCase()));
                }

                @Override
                public String getKey(TradeSkuOrderBean javaBean) {
                    return javaBean.getCategory3Id();
                }
            },
            5 * 60, TimeUnit.SECONDS
        );

        //TODO 18.和category2维度进行关联
        SingleOutputStreamOperator<TradeSkuOrderBean> withCategory2Stream = AsyncDataStream.unorderedWait(
            withCategory3Stream,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category2".toUpperCase()) {
                @Override
                public void join(TradeSkuOrderBean javaBean, JSONObject jsonObj) {
                    javaBean.setCategory2Name(jsonObj.getString("name".toUpperCase()));
                    javaBean.setCategory1Id(jsonObj.getString("category1_id".toUpperCase()));
                }

                @Override
                public String getKey(TradeSkuOrderBean javaBean) {
                    return javaBean.getCategory2Id();
                }
            },
            5 * 60, TimeUnit.SECONDS
        );

        //TODO 19.和category1维度进行关联
        SingleOutputStreamOperator<TradeSkuOrderBean> withCategory1Stream = AsyncDataStream.unorderedWait(
            withCategory2Stream,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category1".toUpperCase()) {
                @Override
                public void join(TradeSkuOrderBean javaBean, JSONObject jsonObj) {
                    javaBean.setCategory1Name(jsonObj.getString("name".toUpperCase()));
                }

                @Override
                public String getKey(TradeSkuOrderBean javaBean) {
                    return javaBean.getCategory1Id();
                }
            },
            5 * 60, TimeUnit.SECONDS
        );

        //TODO 20.将聚合结果写到CK中
        withCategory1Stream.print(">>>");
        withCategory1Stream.addSink(MyClickhouseUtil.getSinkFunction("insert into dws_trade_sku_order_window values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"));

        env.execute();
    }
}
